package com.aispeech.asr.apis.ssc.beans;

import java.net.MalformedURLException;
import java.net.URISyntaxException;
import java.net.URL;
import java.util.function.Function;

import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.annotation.Scope;
import org.springframework.data.mongodb.core.MongoOperations;
import org.springframework.http.MediaType;
import org.springframework.stereotype.Component;
import org.springframework.web.reactive.function.client.WebClient;

import com.aispeech.asr.apis.ssc.constants.Progress;
import com.aispeech.asr.apis.ssc.data.domain.TaskDoc;
import com.aispeech.asr.apis.ssc.data.vo.TaskDocInfo;
import com.aispeech.asr.apis.ssc.event.SscEvent;
import com.aispeech.asr.apis.ssc.logger.BigDataLogger;
import com.aispeech.asr.apis.ssc.service.ProgressManager;
import com.aispeech.asr.comm.data.vo.Result;
import com.aispeech.asr.comm.exceptions.AsrException;

import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import reactor.core.publisher.Mono;

@Slf4j
@Component
@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
public class Completer implements Function<SscEvent, Mono<SscEvent>> {
    @Autowired
    ProgressManager progressManager;
    @Autowired
    MongoOperations mongoOperations;
    @Autowired
    WebClient webClient;
//    @Value("${apis.boss-metering-url}")
//    private String bossMeteringUrl;

    @SneakyThrows
    @Override
    public Mono<SscEvent> apply(SscEvent event) {
        TaskDoc taskDoc = event.getTaskDoc();
        log.info("[ Completer ] start。。。taskId:{}",taskDoc.getId());
        if (taskDoc.getProgress() < Progress.PARAGRAPHED)
            throw new AsrException("need todo");

        String callback = (String) taskDoc.getRequest().getOrDefault("callback", "");

        Mono<SscEvent> mono = Mono.just(event);
        progressManager.finish(taskDoc);

        if (StringUtils.isNotEmpty(callback)) {
            doCallback(event, callback);
        }

        mono = mono.doOnNext(ev -> {
            //reportDuration(taskDoc.getProductId(), duration);
            BigDataLogger.logTask(taskDoc);
        });
        
        log.info("[ Completer ] end。。。taskId:{}",taskDoc.getId());
        
        return mono;
    }

    private void doCallback(SscEvent event, String callback) throws MalformedURLException, URISyntaxException {
        TaskDoc taskDoc = event.getTaskDoc();
       
        Result taskDocInfo = TaskDocInfo.mergeAnnotation(taskDoc);
        
        URL url = new URL(callback);
        
        webClient.post()
                .uri(url.toURI())
                .contentType(MediaType.APPLICATION_JSON)
                .bodyValue(taskDocInfo)
                .retrieve()
                .bodyToMono(String.class)
                .subscribe(
                        resp -> onResponse(event, resp),
                        throwable -> onException(event.getTaskDoc(), callback, throwable)
                );
    }

    private SscEvent onResponse(SscEvent event, String response) {
        log.warn(
                "[ callback ] taskId: {}, productId: {}, message: {}",
                event.getTaskDoc().getId(),
                event.getTaskDoc().getProductId(),
                response
        );

        return event;
    }

    private void onException(TaskDoc taskDoc, String callback, Throwable throwable) {
        log.warn(
                "[ callback ] taskId: {}, productId: {}, callback: {}, caused by: {}",
                taskDoc.getId(),
                taskDoc.getProductId(),
                callback,
                throwable.getMessage());
    }

//    private void reportDuration(String productId, double duration) {
//        if (StringUtils.isAnyEmpty(bossMeteringUrl)) {
//            log.warn("Not found property: apis.boss-metering-url");
//        } else {
//            List<Meter> meters = new ArrayList<>();
//
//            if (duration > 0) {
//                meters.add(Meter.of(MeterType.OFFLINE, DimensionType.PERIOD, productId, duration / 1000.0));
//            }
//
//            webClient
//                    .post()
//                    .uri(bossMeteringUrl)
//                    .contentType(MediaType.APPLICATION_JSON)
//                    .bodyValue(meters)
//                    .retrieve()
//                    .bodyToMono(String.class)
//                    .subscribe(res -> {
//                        log.info("[ MeterSyncer.subscribe ] response: {}", res);
//                    });
//        }
//    }
}
